import process from 'node:process';
import {EventEmitter} from 'node:events';
import stream from 'node:stream';
import {pipeline as streamPipeline} from 'node:stream/promises';
import http from 'node:http';
import https from 'node:https';
import net from 'node:net';
import getStream from 'get-stream';
import test from 'ava';
import delay from 'delay';
import type {Handler} from 'express';
import {pEvent} from 'p-event';
import got, {type RequestError, TimeoutError} from '../source/index.js';
import type {NativeRequestOptions} from '../source/core/options.js';
import timedOut from '../source/core/timed-out.js';
import slowDataStream from './helpers/slow-data-stream.js';
import type {GlobalClock} from './helpers/types.js';
import withServer, {withServerAndFakeTimers, withHttpsServer} from './helpers/with-server.js';

const requestDelay = 800;

const errorMatcher = {
	instanceOf: TimeoutError,
	code: 'ETIMEDOUT',
};

const keepAliveAgent = new http.Agent({
	keepAlive: true,
});

const defaultHandler = (clock: GlobalClock): Handler => (request, response) => {
	request.resume();
	request.on('end', () => {
		clock.tick(requestDelay);
		response.end('OK');
	});
};

const downloadHandler = (clock?: GlobalClock): Handler => (_request, response) => {
	response.writeHead(200, {
		'transfer-encoding': 'chunked',
	});
	response.flushHeaders();

	setImmediate(async () => {
		await streamPipeline(slowDataStream(clock), response);
	});
};

test.serial('timeout option', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));

	await t.throwsAsync(
		got({
			timeout: {
				request: 1,
			},
			retry: {
				limit: 0,
			},
		}),
		{
			...errorMatcher,
			message: 'Timeout awaiting \'request\' for 1ms',
		},
	);
});

test.serial('timeout option as object', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));

	await t.throwsAsync(
		got({
			timeout: {request: 1},
			retry: {
				limit: 0,
			},
		}),
		{
			...errorMatcher,
			message: 'Timeout awaiting \'request\' for 1ms',
		},
	);
});

test.serial('socket timeout', async t => {
	await t.throwsAsync(
		got('https://example.com', {
			timeout: {socket: 1},
			retry: {
				limit: 0,
			},
			request() {
				return https.request('https://example.com', {
					timeout: 0,
				});
			},
		}),
		{
			instanceOf: TimeoutError,
			code: 'ETIMEDOUT',
			message: 'Timeout awaiting \'socket\' for 1ms',
		},
	);
});

test.serial('send timeout', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.post('/', defaultHandler(clock));

	await t.throwsAsync(
		got.post({
			timeout: {send: 1},
			body: new stream.PassThrough(),
			retry: {
				limit: 0,
			},
		}).on('request', request => {
			request.once('socket', socket => {
				socket.once('connect', () => {
					clock.tick(10);
				});
			});
		}),
		{
			...errorMatcher,
			message: 'Timeout awaiting \'send\' for 1ms',
		},
	);
});

test.serial('send timeout (keepalive)', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.post('/', defaultHandler(clock));
	server.get('/prime', (_request, response) => {
		response.end('ok');
	});

	await got('prime', {agent: {http: keepAliveAgent}});

	await t.throwsAsync(
		got.post({
			agent: {
				http: keepAliveAgent,
			},
			timeout: {send: 1},
			retry: {
				limit: 0,
			},
			body: slowDataStream(clock),
		}).on('request', (request: http.ClientRequest) => {
			request.once('socket', socket => {
				t.false(socket.connecting);

				socket.once('connect', () => {
					t.fail('\'connect\' event fired, invalidating test');
				});
			});
		}),
		{
			...errorMatcher,
			message: 'Timeout awaiting \'send\' for 1ms',
		},
	);
});

test.serial('response timeout', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));

	await t.throwsAsync(
		got({
			timeout: {response: 1},
			retry: {
				limit: 0,
			},
		}),
		{
			...errorMatcher,
			message: 'Timeout awaiting \'response\' for 1ms',
		},
	);
});

test.serial('response timeout unaffected by slow upload', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.post('/', defaultHandler(clock));

	await t.notThrowsAsync(got.post({
		retry: {
			limit: 0,
		},
		body: slowDataStream(clock),
	}));
});

test.serial('response timeout unaffected by slow download', withServer, async (t, server, got) => {
	server.get('/', downloadHandler());

	await t.notThrowsAsync(got({
		timeout: {response: 200},
		retry: {
			limit: 0,
		},
	}));

	await delay(100);
});

test.serial('response timeout (keepalive)', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));
	server.get('/prime', (_request, response) => {
		response.end('ok');
	});

	await got('prime', {agent: {http: keepAliveAgent}});

	const request = got({
		agent: {
			http: keepAliveAgent,
		},
		timeout: {response: 1},
		retry: {
			limit: 0,
		},
	}).on('request', (request: http.ClientRequest) => {
		request.once('socket', socket => {
			t.false(socket.connecting);
			socket.once('connect', () => {
				t.fail('\'connect\' event fired, invalidating test');
			});
		});
	});

	await t.throwsAsync(request, {
		...errorMatcher,
		message: 'Timeout awaiting \'response\' for 1ms',
	});
});

test.serial('connect timeout', withServerAndFakeTimers, async (t, _server, got, clock) => {
	await t.throwsAsync(
		got({
			createConnection(options: NativeRequestOptions) {
				const socket = new net.Socket(options as Record<string, unknown> as net.SocketConstructorOpts);
				// @ts-expect-error We know that it is readonly, but we have to test it
				socket.connecting = true;
				setImmediate(() => {
					socket.emit('lookup', null, '127.0.0.1', 4, 'localhost');
				});
				return socket;
			},
			timeout: {connect: 1},
			retry: {
				limit: 0,
			},
		}).on('request', (request: http.ClientRequest) => {
			request.on('socket', () => {
				clock.runAll();
			});
		}),
		{
			...errorMatcher,
			message: 'Timeout awaiting \'connect\' for 1ms',
		},
	);
});

test.serial('connect timeout (ip address)', withServerAndFakeTimers, async (t, _server, _got, clock) => {
	await t.throwsAsync(
		got({
			url: 'http://127.0.0.1',
			createConnection(options: NativeRequestOptions) {
				const socket = new net.Socket(options as Record<string, unknown> as net.SocketConstructorOpts);
				// @ts-expect-error We know that it is readonly, but we have to test it
				socket.connecting = true;
				return socket;
			},
			timeout: {connect: 1},
			retry: {
				limit: 0,
			},
		}).on('request', (request: http.ClientRequest) => {
			request.on('socket', () => {
				clock.runAll();
			});
		}),
		{
			...errorMatcher,
			message: 'Timeout awaiting \'connect\' for 1ms',
		},
	);
});

test.serial('secureConnect timeout', withHttpsServer({}, true), async (t, _server, got, clock) => {
	await t.throwsAsync(
		got({
			createConnection(options: NativeRequestOptions) {
				const socket = new net.Socket(options as Record<string, unknown> as net.SocketConstructorOpts);
				// @ts-expect-error We know that it is readonly, but we have to test it
				socket.connecting = true;
				setImmediate(() => {
					socket.emit('lookup', null, '127.0.0.1', 4, 'localhost');

					setImmediate(() => {
						socket.emit('connect');
					});
				});
				return socket;
			},
			timeout: {secureConnect: 0},
			retry: {
				limit: 0,
			},
		}).on('request', (request: http.ClientRequest) => {
			request.on('socket', () => {
				clock.runAll();
			});
		}),
		{
			...errorMatcher,
			message: 'Timeout awaiting \'secureConnect\' for 0ms',
		},
	);
});

test('secureConnect timeout not breached', withServer, async (t, server, got) => {
	server.get('/', (_request, response) => {
		response.end('ok');
	});

	await t.notThrowsAsync(got({
		timeout: {secureConnect: 200},
		retry: {
			limit: 0,
		},
		https: {
			rejectUnauthorized: false,
		},
	}));
});

test.serial('lookup timeout', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));

	await t.throwsAsync(
		got({
			dnsLookup() {},
			timeout: {lookup: 1},
			retry: {
				limit: 0,
			},
		}).on('request', (request: http.ClientRequest) => {
			request.on('socket', () => {
				clock.runAll();
			});
		}),
		{
			...errorMatcher,
			message: 'Timeout awaiting \'lookup\' for 1ms',
		},
	);
});

test.serial('lookup timeout no error (ip address)', withServerAndFakeTimers, async (t, server, _got, clock) => {
	server.get('/', defaultHandler(clock));

	await t.notThrowsAsync(got({
		url: `http://127.0.0.1:${server.port}`,
		timeout: {lookup: 1},
		retry: {limit: 0},
	}));
});

test.serial('lookup timeout no error (keepalive)', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));
	server.get('/prime', (_request, response) => {
		response.end('ok');
	});

	await got('prime', {agent: {http: keepAliveAgent}});
	await t.notThrowsAsync(got({
		agent: {http: keepAliveAgent},
		timeout: {lookup: 1},
		retry: {limit: 0},
	}).on('request', (request: http.ClientRequest) => {
		request.once('connect', () => {
			t.fail('connect event fired, invalidating test');
		});
	}));

	keepAliveAgent.destroy();
});

test.serial('retries on timeout', withServer, async (t, server, got) => {
	server.get('/', () => {});

	let hasTried = false;
	await t.throwsAsync(got({
		timeout: {
			request: 1,
		},
		retry: {
			calculateDelay() {
				if (hasTried) {
					return 0;
				}

				hasTried = true;
				return 1;
			},
		},
	}), {
		...errorMatcher,
		message: 'Timeout awaiting \'request\' for 1ms',
	});

	t.true(hasTried);
});

test.serial('timeout with streams', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));

	const stream = got.stream({
		timeout: {request: 0},
		retry: {limit: 0},
	});
	await t.throwsAsync(pEvent(stream, 'response'), {code: 'ETIMEDOUT'});
});

test.serial('no error emitted when timeout is not breached (stream)', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));

	const stream = got.stream({
		retry: {
			limit: 0,
		},
		timeout: {
			request: requestDelay * 2,
		},
	});

	await t.notThrowsAsync(getStream(stream));
});

test.serial('no error emitted when timeout is not breached (promise)', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));

	await t.notThrowsAsync(got({
		retry: {
			limit: 0,
		},
		timeout: {
			request: requestDelay * 2,
		},
	}));
});

test.serial('no unhandled `socket hung up` errors', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));

	await t.throwsAsync(
		got({
			retry: {
				limit: 0,
			},
			timeout: {
				request: requestDelay / 2,
			},
		}),
		{instanceOf: TimeoutError},
	);
});

// TODO: use fakeTimers here
test.serial('no unhandled timeout errors', withServer, async (t, _server, got) => {
	await t.throwsAsync(got({
		retry: {limit: 0},
		timeout: {request: 100},
		request(...arguments_) {
			const result = http.request(...arguments_);

			result.once('socket', () => {
				result.socket?.destroy();
			});

			return result;
		},
	}), {message: 'socket hang up'});

	await delay(200);
});

// TODO: use fakeTimers here
test.serial('no unhandled timeout errors #2', withServer, async (t, server, got) => {
	server.get('/', (_request, response) => {
		response.write('Hello world!');
	});

	const gotPromise = got('', {
		timeout: {
			request: 20,
		},
		retry: {
			calculateDelay({computedValue}) {
				if (computedValue) {
					return 10;
				}

				return 0;
			},
			limit: 1,
		},
	});

	await t.throwsAsync(gotPromise, {instanceOf: TimeoutError});

	await delay(100);
});

test.serial('no more timeouts after an error', withServer, async (t, _server, got) => {
	const {setTimeout} = global;
	const {clearTimeout} = global;

	// @ts-expect-error FIXME
	global.setTimeout = (callback, _ms, ...arguments_) => {
		const timeout = {
			isCleared: false,
		};

		process.nextTick(() => {
			if (timeout.isCleared) {
				return;
			}

			callback(...arguments_);
		});

		return timeout;
	};

	global.clearTimeout = timeout => {
		if (timeout) {
			// @ts-expect-error FIXME
			timeout.isCleared = true;
		}
	};

	await t.throwsAsync(got(`http://${Date.now()}.dev`, {
		retry: {limit: 1},
		timeout: {
			lookup: 1,
			connect: 1,
			secureConnect: 1,
			socket: 1,
			response: 1,
			send: 1,
			request: 1,
		},
	}), {instanceOf: TimeoutError});

	await delay(100);

	global.setTimeout = setTimeout;
	global.clearTimeout = clearTimeout;
});

test.serial('socket timeout is canceled on error', withServerAndFakeTimers, async (t, _server, got, clock) => {
	const message = 'oh, snap!';

	const promise = got({
		timeout: {socket: 50},
		retry: {limit: 0},
	}).on('request', (request: http.ClientRequest) => {
		request.destroy(new Error(message));
	});

	await t.throwsAsync(promise, {message});

	// Wait a bit more to check if there are any unhandled errors
	clock.tick(100);
});

test.serial('no memory leak when using socket timeout and keepalive agent', withServerAndFakeTimers, async (t, server, got, clock) => {
	server.get('/', defaultHandler(clock));

	let request: any;

	await got({
		agent: {http: keepAliveAgent},
		timeout: {socket: requestDelay * 2},
	}).on('request', _request => {
		request = _request;
	});

	t.is(request.timeoutCb, null);

	keepAliveAgent.destroy();
});

test('ensure there are no new timeouts after cancelation', t => {
	const emitter = new EventEmitter();
	const socket = new EventEmitter();
	(socket as any).connecting = true;

	timedOut(emitter as http.ClientRequest, {
		connect: 1,
	}, {
		hostname: '127.0.0.1',
	})();

	emitter.emit('socket', socket);
	socket.emit('lookup', null);
	t.is(socket.listenerCount('connect'), 0);
});

test('double calling timedOut has no effect', t => {
	const emitter = new EventEmitter();

	const attach = (): () => void => timedOut(emitter as http.ClientRequest, {
		connect: 1,
	}, {
		hostname: '127.0.0.1',
	});

	attach();
	attach();

	t.is(emitter.listenerCount('socket'), 1);
});

// Regression test for issue #857: Fast DNS lookups (e.g., from cache) that complete synchronously
// used to incorrectly trigger lookup timeouts because Got couldn't attach the 'lookup' event listener
// in time. The fix in commit 4faf5c7 checks if the lookup has already completed (socket.address() is
// defined) before setting a lookup timeout, preventing false timeout errors on cached/fast lookups.
test.serial('doesn\'t throw on early lookup', withServerAndFakeTimers, async (t, server, got) => {
	server.get('/', (_request, response) => {
		response.end('ok');
	});

	await t.notThrowsAsync(got('', {
		timeout: {
			lookup: 1,
		},
		retry: {limit: 0},
		dnsLookup(_hostname, options, callback) {
			if (typeof options === 'function') {
				callback = options;
				// Call with default (non-all) signature
				callback(null, '127.0.0.1', 4);
			} else if (options.all) {
				// When options.all is true, callback expects an array of address objects
				callback(null, [{address: '127.0.0.1', family: 4}]);
			} else {
				callback(null, '127.0.0.1', 4);
			}
		},
	}));
});

// TODO: use fakeTimers here
test.serial('no unhandled `Premature close` error', withServer, async (t, server, got) => {
	server.get('/', async (_request, response) => {
		response.write('hello');
	});

	await t.throwsAsync(got({
		timeout: {request: 10},
		retry: {limit: 0},
	}), {message: 'Timeout awaiting \'request\' for 10ms'});

	await delay(20);
});

// TODO: use fakeTimers here
test.serial('`read` timeout - promise', withServer, async (t, server, got) => {
	server.get('/', (_request, response) => {
		response.write('o');
	});

	await t.throwsAsync(got({
		timeout: {
			read: 10,
		},
		retry: {
			limit: 0,
		},
	}), {message: 'Timeout awaiting \'read\' for 10ms'});
});

// TODO: use fakeTimers here
test.serial('`read` timeout - stream', withServer, async (t, server, got) => {
	t.timeout(100);

	server.get('/', (_request, response) => {
		response.end('ok');
	});

	const stream = got.stream({
		timeout: {
			read: 10,
		},
	});

	await t.throwsAsync(pEvent(stream, 'end'), {message: 'Timeout awaiting \'read\' for 10ms'});
});

// TODO: use fakeTimers here
test.serial('cancelling the request removes timeouts', withServer, async (t, server, got) => {
	server.get('/', (_request, response) => {
		response.write('hello');
	});

	const controller = new AbortController();

	const promise = got({
		signal: controller.signal,
		timeout: {
			request: 500,
		},
		retry: {
			limit: 0,
		},
	}).on('downloadProgress', () => {
		controller.abort();
	}).on('request', request => {
		request.on('error', error => {
			if (error.message === 'Timeout awaiting \'request\' for 500ms') {
				t.fail(error.message);
			}
		});
	});

	await t.throwsAsync(promise, {message: 'This operation was aborted.'});

	await delay(1000);
});

test.serial('timeouts are emitted ASAP', withServer, async (t, server, got) => {
	server.get('/test', () => {
		// Never respond - forces a timeout
	});

	const timeout = 500;
	const marginOfError = process.env.CI ? 200 : 100;
	const startTime = Date.now();

	const error = await t.throwsAsync<TimeoutError>(got('test', {
		retry: {
			limit: 0,
		},
		timeout: {
			request: timeout,
		},
	}), {instanceOf: TimeoutError});

	const elapsed = Date.now() - startTime;
	t.true(elapsed < (timeout + marginOfError), `Expected timeout ${elapsed}ms to be less than ${timeout + marginOfError}ms`);
	t.true(error!.timings.phases.total! < (timeout + marginOfError));
});

test('http2 timeout', async t => {
	const error = await t.throwsAsync<RequestError>(got('https://123.123.123.123', {
		timeout: {
			request: 1,
		},
		http2: true,
		retry: {
			calculateDelay: ({computedValue}) => computedValue ? 1 : 0,
		},
	}));

	t.true(error?.code === 'ETIMEDOUT' || error?.code === 'EUNSUPPORTED', error?.stack);
});

// Reproduces the memory leak reported in https://github.com/sindresorhus/got/issues/2351
test.serial('no memory leak when using http2 with socket timeout and connection reuse', withHttpsServer(), async (t, server, got) => {
	const {default: http2wrapper} = await import('http2-wrapper');
	const customAgent = new http2wrapper.Agent();

	server.get('/:id', (_request, response) => {
		response.end('ok');
	});

	const requestCount = 15; // Make concurrent requests to accumulate listeners
	let sharedSocket: any;
	let maxListenerCount = 0;

	// Make requests that overlap in time to reproduce the listener accumulation
	server.get('/slow/:id', async (_request, response) => {
		// Slow response to keep requests active simultaneously
		await delay(200);
		response.end('ok');
	});

	// Start many concurrent requests with the same agent to reuse the connection
	const promises = [];
	const handleRequest = (request: any) => {
		// Track socket and check listener count DURING requests
		request.once('socket', (socket: any) => {
			if (!sharedSocket) {
				sharedSocket = socket;
			}

			// Check how many listeners are on the socket while requests are active
			const currentCount = socket.listenerCount('timeout');
			if (currentCount > maxListenerCount) {
				maxListenerCount = currentCount;
			}
		});
	};

	for (let i = 0; i < requestCount; i++) {
		const promise = got(`slow/${i}`, {
			http2: true,
			agent: {http2: customAgent},
			timeout: {socket: 30_000},
			https: {rejectUnauthorized: false},
		}).on('request', handleRequest);
		promises.push(promise);
	}

	// Wait for all concurrent requests to finish
	await Promise.all(promises);

	// The bug: setTimeout(0) doesn't remove timeout listeners, so with HTTP/2
	// connection reuse and concurrent requests, listeners accumulate on the shared socket
	// The fix: removeAllListeners('timeout') properly cleans up

	t.truthy(sharedSocket, 'Should have a socket');

	// With the bug (setTimeout(0)), maxListenerCount would be >> 1 (grows with concurrent requests)
	// With the fix (removeAllListeners), maxListenerCount should be 0 or 1
	t.true(maxListenerCount <= 1, `Socket peaked at ${maxListenerCount} timeout listeners (expected ≤ 1)`);

	customAgent.destroy();
});
